package com.weds.data_sync.service.impl;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.annotation.JSONField;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.weds.data_sync.advice.exception.BizCodeException;
import com.weds.data_sync.constant.BizCodeEnum;
import com.weds.data_sync.constant.ScriptServiceConstant;
import com.weds.data_sync.event.ExecutorPlanErrorEvent;
import com.weds.data_sync.event.ExecutorPlanFinishEvent;
import com.weds.data_sync.mapper.DataSyncPlanRecordMapper;
import com.weds.data_sync.pojo.dto.ScriptRequestDto;
import com.weds.data_sync.pojo.dto.ScriptResponseDto;
import com.weds.data_sync.pojo.entity.DataSyncPlanRecordDO;
import com.weds.data_sync.pojo.entity.DataSyncScriptDO;
import com.weds.data_sync.pojo.entity.DataSyncScriptPlanDO;
import com.weds.data_sync.service.ExecutePlan;
import com.weds.data_sync.service.ScriptService;
import com.weds.data_sync.util.IDUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.util.StringUtils;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * @Description 执行方案实现类
 * @Author lihw
 * @Date 2024/1/2 16:47
 */
@Slf4j
@Data
public class ExecutePlanImpl implements ExecutePlan {

    /**
     * 待执行执行计划队列
     */
    private static final List<ExecutePlanImpl> planList = new LinkedList<>();

    /**
     * 执行中的执行计划队列
     * key: sessionId, value: 执行计划
     */
    private static final Map<String, ExecutePlan> executePlanMap = new HashMap<>();

    private static final Integer planListMaxLength = 100;

    /**
     * 超时时间 2 天
     */
    private static final long timeout = 1000L * 60 * 60 * 24 * 2;

    private DataSyncScriptPlanDO dataSyncScriptPlanDO;

    private List<DataSyncScriptDO> scriptDOList;

    private Integer scriptLength;

    private Integer executeIndex;

    @JSONField(serialize = false)
    private RestTemplate restTemplate;

    private Object finalResult;

    private String sessionId;

    private ScriptRequestDto.DBInfo dbInfo;

    @JSONField(serialize = false)
    private RedisTemplate<Object, Object> redisTemplate;

    private String redisPlanLockKeyTemplate = "sync:plan:lock:0:%s";

    private String redisPlanLockKey;

    private Long redisPlanLockTtl = 1000L * 60L * 60L * 2L;

    private ApplicationEventPublisher applicationEventPublisher;

    // private DataSyncScriptPlanDTO dataSyncScriptPlanDTO;

    private ScriptService scriptService;

    // 执行方式 1 同步 2 异步
    private Integer executeType;

    private WebClient webClient;

    private LocalDateTime createTime = LocalDateTime.now();

    private DataSyncPlanRecordMapper planRecordMapper;


    public ExecutePlanImpl(DataSyncScriptPlanDO dataSyncScriptPlanDO,
                           List<DataSyncScriptDO> scriptDOList,
                           ScriptRequestDto.DBInfo dbInfo,
                           RedisTemplate redisTemplate,
                           ApplicationEventPublisher applicationEventPublisher,
                           ScriptService scriptService,
                           DataSyncPlanRecordMapper planRecordMapper) {
        this.dataSyncScriptPlanDO = dataSyncScriptPlanDO;
        this.scriptDOList = scriptDOList;
        this.scriptLength = scriptDOList.size();
        this.executeIndex = 0;
        this.restTemplate = new RestTemplate();
        this.dbInfo = dbInfo;
        this.sessionId = IDUtil.geneSnowFlakeID().toString();
        this.redisTemplate = redisTemplate;
        this.applicationEventPublisher = applicationEventPublisher;
        this.scriptService = scriptService;
        this.planRecordMapper = planRecordMapper;

        // 添加执行队列
        executePlanMap.put(sessionId, this);
        clearTimeoutPlan(planRecordMapper);
    }

    @Override
    public Object executeScript(Object... params) {
        Object previousResult = null;
        if (params.length > 0) {
            previousResult = params[0];
        }

        if (Objects.isNull(executeIndex)) {
            executeIndex = 0;
        }

        if (executeType != null && executeType != 1) {
            return null;
        }

        if (executeIndex.equals(0)) {

            String planLockKey = String.format(redisPlanLockKeyTemplate, dataSyncScriptPlanDO.getPlanNo());
            redisPlanLockKey = planLockKey;
            Boolean lockFlag = redisTemplate.opsForValue().setIfAbsent(planLockKey, 1, redisPlanLockTtl,
                    TimeUnit.MILLISECONDS);
            if (lockFlag.equals(Boolean.FALSE)) {
                log.info("有同类型的执行方案正在执行");
                if (planList.size() < planListMaxLength) {
                    planList.add(this);
                }
                return null;
            }
            // 设置执行方式
            executeType = 1;
        }

        if (executeIndex >= scriptLength) {
            log.info("未在执行方案中找到改脚本，脚本索引越界。");
            executePlanMap.remove(sessionId);
            return null;
        }

        try {
            DataSyncScriptDO dataSyncScriptDO = scriptDOList.get(executeIndex);
            // 执行之前执行计划存入pg
            catchPlanRecord(ScriptServiceConstant.PlanRecordStatus.EXECUTING);

            ScriptResponseDto scriptResponseDto = null;
            if (dataSyncScriptDO.getRunType().equals(1)) {
                scriptResponseDto = runScriptByRest(dataSyncScriptDO, previousResult);
            } else {
                // 本地执行
                scriptService.runScriptNative(dataSyncScriptDO);
            }

            executeIndex++;
            if (executeIndex >= scriptLength) {
                finalResult = scriptResponseDto;
                executePlanFinish();

                return scriptResponseDto;
            } else {
                // 远程执行
                if (dataSyncScriptDO.getRunType().equals(1)) {
                    if (!Objects.isNull(scriptResponseDto) && scriptResponseDto.getCode().equals(0)) {
                        return this.executeScript(scriptResponseDto.getData());
                    } else {
                        log.info("脚本执行出错,执行方案: {}, 脚本索引: {}, 错误信息: {}", dataSyncScriptPlanDO, executeIndex,
                                scriptResponseDto.getMsg());

                        executePlanError(new RuntimeException(scriptResponseDto.getMsg()));

                        return null;
                    }
                } else {
                    // 本地执行
                    return this.executeScript();
                }

            }
        } catch (Exception e) {
            e.printStackTrace();

            executePlanError(e);

            throw new BizCodeException(BizCodeEnum.SCRIPT_EXEC_ERROR);
        }
    }

    @Override
    public void executeScriptAsync(WebClient webClient, Object... params) {
        Object previousResult = null;
        if (params.length > 0) {
            previousResult = params[0];
        }

        if (Objects.isNull(executeIndex)) {
            executeIndex = 0;
        }

        if (executeType != null && executeType != 2) {
            return;
        }

        if (executeIndex.equals(0)) {

            String planLockKey = String.format(redisPlanLockKeyTemplate, dataSyncScriptPlanDO.getPlanNo());
            redisPlanLockKey = planLockKey;
            Boolean lockFlag = redisTemplate.opsForValue().setIfAbsent(planLockKey, 1, redisPlanLockTtl,
                    TimeUnit.MILLISECONDS);
            if (lockFlag.equals(Boolean.FALSE)) {
                log.info("有同类型的执行方案正在执行");
                if (planList.size() < planListMaxLength) {
                    planList.add(this);
                }
                return;
            }
            // 设置执行方式
            executeType = 2;

        }

        if (executeIndex >= scriptLength) {
            executePlanMap.remove(sessionId);
            return;
        }

        try {
            DataSyncScriptDO dataSyncScriptDO = scriptDOList.get(executeIndex);
            // 缓存执行计划
            catchPlanRecord(ScriptServiceConstant.PlanRecordStatus.EXECUTING);

            executeIndex++;
            if (dataSyncScriptDO.getRunType().equals(1)) {

                runScriptByRestAsync(dataSyncScriptDO, webClient, previousResult);

            } else {
                // 本地执行
                scriptService.runScriptNative(dataSyncScriptDO);
                if (executeIndex >= scriptLength) {
                    executePlanFinish();
                    return;
                } else {
                    this.executeScriptAsync(webClient);
                }

            }


        } catch (Exception e) {
            e.printStackTrace();

            executePlanError(e);

            throw new BizCodeException(BizCodeEnum.SCRIPT_EXEC_ERROR);
        }

    }

    /**
     * rest执行脚本
     *
     * @param dataSyncScriptDO
     * @param previousResult
     * @return
     */
    private ScriptResponseDto runScriptByRest(DataSyncScriptDO dataSyncScriptDO, Object previousResult) {

        // 创建请求头并设置Content-Type为application/json
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON);

        ScriptRequestDto requestParam = ScriptRequestDto.builder()
                .dir(dataSyncScriptDO.getModulePath())
                .module(dataSyncScriptDO.getModuleName())
                .className(dataSyncScriptDO.getClassName())
                .method(dataSyncScriptDO.getFuncName())
                .dbInfo(dbInfo)
                .ackId(sessionId)
                .data(previousResult)
                .build();
        HttpEntity<ScriptRequestDto> httpEntity = new HttpEntity<>(requestParam, headers);
        ResponseEntity<ScriptResponseDto> responseEntity = restTemplate.postForEntity(
                ScriptServiceConstant.SCRIPT_REQUEST_HOST + ScriptServiceConstant.SCRIPT_REQUEST_API,
                httpEntity, ScriptResponseDto.class);
        return responseEntity.getBody();

    }

    /**
     * 异步远程执行脚本
     *
     * @param dataSyncScriptDO
     * @param previousResult
     */
    private void runScriptByRestAsync(DataSyncScriptDO dataSyncScriptDO, WebClient webClient, Object previousResult) {

        ScriptRequestDto requestParam = ScriptRequestDto.builder()
                .dir(dataSyncScriptDO.getModulePath())
                .module(dataSyncScriptDO.getModuleName())
                .className(dataSyncScriptDO.getClassName())
                .method(dataSyncScriptDO.getFuncName())
                .dbInfo(dbInfo)
                .ackId(sessionId)
                .data(previousResult)
                .build();

        Mono<ScriptResponseDto> responseDtoMono = webClient
                .post()
                .uri(ScriptServiceConstant.SCRIPT_REQUEST_API)
                .bodyValue(requestParam)
                .retrieve()
                .bodyToMono(ScriptResponseDto.class);
        responseDtoMono.subscribe((responseDto) -> {
            if (responseDto.getCode() == 0) {
                if (executeIndex >= scriptLength) {
                    if (StringUtils.hasLength(redisPlanLockKey)) {
                        redisTemplate.delete(redisPlanLockKey);
                    }
                    finalResult = responseDto;

                    executePlanFinish();

                } else {
                    this.executeScriptAsync(webClient, responseDto.getData());
                }

            } else {

                executePlanError(new RuntimeException(responseDto.getMsg()));

                throw new RuntimeException(responseDto.getMsg());
            }
        }, (error) -> {
            log.info("脚本执行错误: {}", error);
            RuntimeException runtimeException = new RuntimeException(error);
            executePlanError(runtimeException);
            throw runtimeException;
        });
    }

    /**
     * 清空过期执行计划
     *
     * @param planRecordMapper
     */
    public static void clearTimeoutPlan(DataSyncPlanRecordMapper planRecordMapper) {
        if (executePlanMap.size() <= 0) {
            return;
        }
        List<String> timeoutKeys = new ArrayList<>();

        LocalDateTime now = LocalDateTime.now();
        executePlanMap.values().forEach(item -> {
            ExecutePlanImpl executePlan = (ExecutePlanImpl) item;
            LocalDateTime createTimeTmp = executePlan.getCreateTime();
            Duration between = Duration.between(createTimeTmp, now);
            long millis = between.toMillis();
            if (millis > timeout) {
                DataSyncPlanRecordDO planRecordDO = planRecordMapper.selectOne(new QueryWrapper<DataSyncPlanRecordDO>()
                        .eq("planSessionId", executePlan.getSessionId()));
                int retryCount = 0;
                if (!Objects.isNull(planRecordDO)) {
                    retryCount = planRecordDO.getRetryCount();
                }
                retryCount++;
                DataSyncPlanRecordDO planRecordOperate = DataSyncPlanRecordDO.builder()
                        .planId(executePlan.getDataSyncScriptPlanDO().getId())
                        .execTime(now)
                        .planSessionId(executePlan.getSessionId())
                        .planData(JSON.toJSONString(executePlan))
                        .isSuccess(ScriptServiceConstant.PlanRecordStatus.TIMEOUT)
                        .retryCount(retryCount)
                        .build();
                // 插入
                if (Objects.isNull(planRecordDO)) {
                    // 超时
                    int insert = planRecordMapper.insert(planRecordOperate);
                    log.info("插入{}条方案执行记录, data: {}", insert, planRecordOperate);
                } else {
                    planRecordOperate.setId(planRecordDO.getId());
                    int updateRow = planRecordMapper.updateById(planRecordOperate);
                    log.info("插入{}条方案执行记录, data: {}", updateRow, planRecordOperate);
                }


                timeoutKeys.add(executePlan.getSessionId());
            }
        });

        // 清空过期执行计划
        for (int i = 0; i < timeoutKeys.size(); i++) {
            executePlanMap.remove(timeoutKeys.get(i));
        }
    }

    /**
     * 缓存执行计划
     *
     * @param status
     * @return
     */
    public Integer catchPlanRecord(Integer status) {
        DataSyncPlanRecordDO planRecordDO = planRecordMapper.selectOne(new QueryWrapper<DataSyncPlanRecordDO>()
                .eq("plan_session_id", this.getSessionId()));
        int retryCount = 0;
        if (!Objects.isNull(planRecordDO)) {
            retryCount = planRecordDO.getRetryCount();
        }
        retryCount++;
        LocalDateTime now = LocalDateTime.now();
        DataSyncPlanRecordDO planRecordOperate = DataSyncPlanRecordDO.builder()
                .planId(this.getDataSyncScriptPlanDO().getId())
                .execTime(now)
                .planSessionId(this.getSessionId())
                .planData(JSON.toJSONString(this))
                .isSuccess(status)
                .retryCount(retryCount)
                .build();
        if (Objects.isNull(planRecordDO)) {
            int insert = planRecordMapper.insert(planRecordOperate);
            log.info("插入{}条方案执行记录, data: {}", insert, planRecordOperate);
            return insert;
        } else {
            planRecordOperate.setId(planRecordDO.getId());
            int updateRow = planRecordMapper.updateById(planRecordOperate);
            log.info("更新{}天方案执行记录, data: {}", updateRow, planRecordOperate);
            return updateRow;
        }
    }

    /**
     * 执行计划执行完成回调
     */
    public void executePlanFinish() {
        if (StringUtils.hasLength(redisPlanLockKey)) {
            redisTemplate.delete(redisPlanLockKey);
        }

        DataSyncPlanRecordDO planRecordDO = planRecordMapper.selectOne(new QueryWrapper<DataSyncPlanRecordDO>()
                .eq("plan_session_id", sessionId)
                .eq("is_success", ScriptServiceConstant.PlanRecordStatus.EXECUTING));
        if (Objects.isNull(planRecordDO)) {
            log.info("执行计划finish回调:::未找到正在执行的计划。data: {}", this);
            return;
        }

        // 发布触发事件
        ExecutorPlanFinishEvent executorPlanFinishEvent = new ExecutorPlanFinishEvent(this);
        applicationEventPublisher.publishEvent(executorPlanFinishEvent);

        ExecutePlanImpl removeEle = (ExecutePlanImpl) executePlanMap.remove(sessionId);

        // 更新执行记录状态
        int update = planRecordMapper.update(new UpdateWrapper<DataSyncPlanRecordDO>()
                .eq("plan_session_id", sessionId)
                .set("is_success", ScriptServiceConstant.PlanRecordStatus.SUCCESS));
        log.info("执行计划完成回调,更新{}条执行计划执行状态，data: {}", update, removeEle);
    }

    /**
     * 执行计划执行错误
     */
    public void executePlanError(Exception e) {
        if (StringUtils.hasLength(redisPlanLockKey)) {
            redisTemplate.delete(redisPlanLockKey);
        }

        DataSyncPlanRecordDO planRecordDO = planRecordMapper.selectOne(new QueryWrapper<DataSyncPlanRecordDO>()
                .eq("plan_session_id", sessionId)
                .eq("is_success", ScriptServiceConstant.PlanRecordStatus.EXECUTING));
        if (Objects.isNull(planRecordDO)) {
            log.info("执行计划error回调:::未找到正在执行的计划。data: {}", this);
            return;
        }

        ExecutePlan removeEle = executePlanMap.remove(sessionId);

        // 发布触发事件
        ExecutorPlanErrorEvent errorEvent = new ExecutorPlanErrorEvent(this, executeIndex, e.getMessage());
        applicationEventPublisher.publishEvent(errorEvent);

        // 更新执行记录状态
        int update = planRecordMapper.update(new UpdateWrapper<DataSyncPlanRecordDO>()
                .eq("plan_session_id", sessionId)
                .set("is_success", ScriptServiceConstant.PlanRecordStatus.FAIL));
        log.info("执行计划执行出错,更新{}条执行计划执行状态，data: {}", update, removeEle);

    }
}
